package com.wqd.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.wqd.common.constants.ScheduleConstants;
import com.wqd.common.redis.RedisCacheService;
import com.wqd.model.schedule.dto.Task;
import com.wqd.model.schedule.pojos.Taskinfo;
import com.wqd.model.schedule.pojos.TaskinfoLogs;
import com.wqd.schedule.mapper.TaskinfoLogsMapper;
import com.wqd.schedule.mapper.TaskinfoMapper;
import com.wqd.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Calendar;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

@Service
@Transactional
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 添加延迟任务(有返回值)
     * @param task 任务对象
     * @return 任务id
     */
    @Override
    public Long addTask(Task task) {

        //添加到线程池
        Future<Long> future = threadPoolTaskExecutor.submit(new Callable<Long>() {
            @Override
            public Long call() throws Exception {
                //如果任务添加数据库成功,那就加入到Redis
                if (addTaskToDB(task)) {
                    addTaskToRedis(task);
                }
                log.info("活动线程: " + threadPoolTaskExecutor.getActiveCount());
                //返回延时任务id
                return task.getTaskId();
            }
        });
        try {
            return future.get();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }


    @Autowired
    private RedisCacheService redisCacheService;

    /**
     * 封装方法: 把Task添加到Redis中
     *
     * @param task
     */
    private void addTaskToRedis(Task task) {
        //把类型和优先级拼一起做KeyName
        String key = task.getTaskType() + "_" + task.getPriority();
        //如果执行时间小于当前时间则加入list
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            redisCacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        }
        //如果执行时间大于当前时间,小于预设时间,就加到zset中
        Calendar calendar = Calendar.getInstance();
        //当前时间加五分钟
        calendar.add(Calendar.MINUTE, 5);
        //转换成毫秒值
        long timeInMillis = calendar.getTimeInMillis();
        if (task.getExecuteTime() >= System.currentTimeMillis() && task.getExecuteTime() <= timeInMillis) {
            //时间作为score
            redisCacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
        }
    }


    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;


    /**
     * 封装方法: 用来存储延迟任务到数据库中
     *
     * @param task
     * @return
     */
    private boolean addTaskToDB(Task task) {
        boolean flag = false;

        try {
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task, taskinfo);
            //设置执行时间
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            //插入数据,获取id,设置Task的id
            taskinfoMapper.insert(taskinfo);
            task.setTaskId(taskinfo.getTaskId());
            //保存日志
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo, taskinfoLogs);
            //设置版本号
            taskinfoLogs.setVersion(1);
            //初始化状态
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskinfoLogsMapper.insert(taskinfoLogs);
            flag = true;
        } catch (BeansException e) {
            e.printStackTrace();
        }

        return flag;
    }


    /**
     * 删除任务
     *
     * @param taskId 任务id
     * @return
     */
    @Override
    public boolean cancelTask(long taskId) {
        boolean flag = false;
        //删除任务,更新日志   已执行状态
        Task task = updateDb(taskId, ScheduleConstants.EXECUTED);
        //删除redis的数据
        if (task != null) {
            removeTaskFromRedis(task);
            flag = true;
        }
        return flag;
    }


    /**
     * 封装方法: 执行后删除redis任务
     *
     * @param task
     */
    private void removeTaskFromRedis(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            //index=0, 删除所有值等于value的元素;
            redisCacheService.lRemove(ScheduleConstants.TOPIC + key, 0, JSON.toJSONString(task));
        } else {
            redisCacheService.zRemove(ScheduleConstants.FUTURE + key, JSON.toJSONString(task));
        }
    }

    /**
     * 封装方法: 用来删除已执行任务的信息,以及更新日志
     *
     * @param taskId
     * @param status
     * @return
     */
    private Task updateDb(long taskId, int status) {
        Task task = null;
        try {
            //删除任务
            taskinfoMapper.deleteById(taskId);
            //获取任务日志
            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
            taskinfoLogs.setStatus(status);
            taskinfoLogsMapper.updateById(taskinfoLogs);
            task = new Task();
            BeanUtils.copyProperties(taskinfoLogs, task);
            task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        } catch (Exception e) {
            log.error("task cancel exception taskid={}", taskId);
            e.printStackTrace();
        }
        return task;
    }


    /**
     * 消费任务,返回Task对象
     * 拉取任务一定是从list拉去的
     * @param type
     * @param priority
     * @return
     */
    @Override
    public Task pullTask(int type, int priority) {
        Future<Task> future = threadPoolTaskExecutor.submit(new Callable<Task>() {
            @Override
            public Task call() throws Exception {
                Task task = null;
                try {
                    String key = type + "_" + priority;
                    //移除并获取列表最后一个元素
                    String taskJson = redisCacheService.lRightPop(ScheduleConstants.TOPIC + key);
                    if (StringUtils.isNotBlank(taskJson)) {
                        task = JSON.parseObject(taskJson, Task.class);
                        //更新数据库
                        updateDb(task.getTaskId(), ScheduleConstants.EXECUTED);
                    }
                } catch (Exception e) {
                    log.error("poll task exception");
                    e.printStackTrace();
                }
                return task;
            }
        });
        try {
            return future.get();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }


    /**
     * 定时任务,每分钟执行一次,将符合要求的zset数据转到list
     */
    @Scheduled(cron = "0 */1 * * * ?")
    @Override
    public void refresh() {

        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //分布式锁
                String token = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);

                if (StringUtils.isNotBlank(token)) {
                    log.info("未来数据定时任务执行......");
                    // 获取所有未来数据集合的key值
                    Set<String> futureKey = redisCacheService.scan(ScheduleConstants.FUTURE + "*");
                    for (String key : futureKey) {
                        //获取key里面的数据
                        Set<String> tasks = redisCacheService.zRangeByScore(key, 0, System.currentTimeMillis());
                        if (!tasks.isEmpty()) {
                            //如果不为空
                            //获取topickey
                            String topicKey = ScheduleConstants.TOPIC + key.split(ScheduleConstants.FUTURE)[1];
                            //添加到消费队列(list)中
                            redisCacheService.refreshWithPipeline(key, topicKey, tasks);
                            System.out.println("成功的将" + key + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
                        }
                    }
                }
            }
        });
    }
}
